package logrustash

import (
	"bytes"
	"encoding/json"
	"errors"
	"github.com/elastic/go-elasticsearch"
	"github.com/sirupsen/logrus"
	"log"
	"net"
	"net/http"
	"strconv"
)

// ElasticsearchConf 配置
type ElasticsearchConf struct {
	Hosts              []string
	Number_of_shards   int
	Number_of_replicas int
}

// LogstashConf 配置
type LogstashConf struct {
	Host      string
	Port      int
	Transport string
}

// KibanaConf 配置
type KibanaConf struct {
	Host string
	Port int
}

// LogELK 封装elk
type LogELK struct {
	es_conf       ElasticsearchConf
	logstash_conf LogstashConf
	kibana_conf   KibanaConf
	platform      string
	check_init    int
	_log          *logrus.Logger
}

func NewELKLog(es_conf ElasticsearchConf, logstash_conf LogstashConf, kibana_conf KibanaConf) *LogELK {
	return &LogELK{
		es_conf:       es_conf,
		logstash_conf: logstash_conf,
		kibana_conf:   kibana_conf,
		check_init:    0,
	}
}

func (self *LogELK) Init(platform string) *LogELK {
	if self.es_conf.Number_of_shards == 0 {
		self.es_conf.Number_of_shards = 1
	}
	self.platform = platform
	self.__create_mapping__()
	self._log = logrus.New()
	self._log.SetLevel(logrus.InfoLevel)
	self.check_init = 1
	return self

}

func (self *LogELK) SyncLogger() (*logrus.Logger, net.Conn) {
	if _hook, conn, err := self._hook(false); err != nil {
		logrus.Error(err)
		return nil, nil
	} else {
		self._log.Hooks.Add(_hook)
		return self._log, conn
	}
}

func (self *LogELK) AsyncLogger() (*logrus.Logger, net.Conn) {
	if _hook, conn, err := self._hook(true); err != nil {
		logrus.Error(err)
		return nil, nil
	} else {
		self._log.Hooks.Add(_hook)
		return self._log, conn
	}
}

func (self *LogELK) _hook(async bool) (Hook, net.Conn, error) {
	self.__check__()
	formatter := DefaultFormatter(logrus.Fields{
		"extra": logrus.Fields{
			"logger_name": self.platform,
		},
	})
	if self.logstash_conf.Transport == "http" {
		url := "http://" + self.logstash_conf.Host + ":" + strconv.Itoa(self.logstash_conf.Port)
		return New(nil, formatter, url, false), nil, nil
	} else if self.logstash_conf.Transport == "tcp" {
		if conn, err := net.Dial("tcp", self.logstash_conf.Host+":"+strconv.Itoa(self.logstash_conf.Port)); err != nil {
			logrus.Error(err)
			return Hook{}, conn, err
		} else {
			return New(conn, formatter, "", false), conn, nil
		}

	} else if self.logstash_conf.Transport == "udp" {
		if conn, err := net.Dial("udp", self.logstash_conf.Host+":"+strconv.Itoa(self.logstash_conf.Port)); err != nil {
			logrus.Error(err)
			return Hook{}, conn, err
		} else {
			return New(conn, formatter, "", false), conn, nil
		}

	} else {
		return Hook{}, nil, errors.New("仅仅支持http/tcp/udp")
	}
}

func (self *LogELK) __check__() {
	if self.check_init != 1 {
		panic("请先执行init()")
	}
}

func (self *LogELK) __create_mapping__() {
	cfg := elasticsearch.Config{
		Addresses: self.es_conf.Hosts,
	}
	if es, err := elasticsearch.NewClient(cfg); err != nil {
		logrus.Error(err)
		return
	} else {
		if _exist, err := es.Indices.Exists([]string{self.platform}); err != nil {
			logrus.Error(err)
			return
		} else {
			statusCode := _exist.StatusCode
			if statusCode == 200 {
				// pass
			} else {
				// create index
				mapping := map[string]interface{}{
					"settings": map[string]interface{}{
						"number_of_shards":   self.es_conf.Number_of_shards,
						"number_of_replicas": self.es_conf.Number_of_replicas,
					},
					"mappings": map[string]interface{}{
						"properties": map[string]interface{}{
							"@timestamp": map[string]interface{}{
								"type": "date",
							},
						},
					},
				}
				var buf bytes.Buffer
				if err := json.NewEncoder(&buf).Encode(mapping); err != nil {
					log.Fatalf("Error encoding mapping: %s", err)
				}
				if _, err := es.Indices.Create(self.platform, es.Indices.Create.WithBody(&buf)); err != nil {
					logrus.Error(err)
					return
				} else {
					self.__create_index_pattern__()
				}
			}
		}
	}
}

func (self *LogELK) __create_index_pattern__() {
	url := "http://" + self.kibana_conf.Host + ":" + strconv.Itoa(self.kibana_conf.Port) + "/api/saved_objects/index-pattern"
	body := map[string]interface{}{
		"attributes": map[string]interface{}{
			"title":         self.platform,
			"timeFieldName": "@timestamp",
		},
	}
	var buf bytes.Buffer
	if err := json.NewEncoder(&buf).Encode(body); err != nil {
		log.Fatalf("Error encoding body: %s", err)
	}
	client := &http.Client{}
	request, _ := http.NewRequest("POST", url, &buf)
	request.Header.Add("Content-Type", "application/json")
	request.Header.Add("kbn-xsrf", "true")
	if res, err := client.Do(request); err != nil {
		logrus.Error(err)
		return
	} else {
		if res.StatusCode != 200 {
			defer res.Body.Close()
			logrus.Error(res.Body)
		}
	}
}
